// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#pragma once

#include <cstdlib>
#include <deque>
#include <functional>
#include <iostream>
#include <map>
#include <memory>
#include <set>
#include <string>
#include <utility>
#include <tuple>
#include <vector>
#include <optional>

#include "coin2/base/config/app_config.h"
#include "coin2/base/log.h"
#include "coin2/base/time.h"
#include "coin2/exchange/factory/feed/request_generator_factory.h"
#include "coin2/exchange/util/rate_limit.h"
#include "cc/coin2/exchange/base/executor/topic_publisher.h"
#include "coin2/flow/symbol_group.h"
#include "coin2/exchange/base/feed/ws/fix_connection.h"

namespace coin2::exchange::base::feed {

using coin2::exchange::feed::WebsocketConfig;

class FixSubscriber : public std::enable_shared_from_this<FixSubscriber>,
                     public api_base::BaseRawAsyncSubscriber {
 public:
  FixSubscriber(
      const FeedSubscriptionSpec& spec,
      const std::vector<WsConnectionConfig>& connection_configs,
      api_base::LiveAsyncExecutor* flow_sub)
      : BaseRawAsyncSubscriber(flow_sub),
        spec_(spec),
        connection_configs_(connection_configs) {
  }

  void RegisterPubTopic() override { flow_sub_->distributor()->Publish(spec_.mea.SnakeString()); }

  void SubscribeAsync() override {
    Subscribe();
  }

  void Subscribe() {
    for (auto& config : connection_configs_) {
      std::shared_ptr<FixConnection> con = FixConnection::NewFixConnection(flow_sub_, config);
      con->Start();
      connections_.push_back(con);
    }
  }

  void GetHealthReport(int indent, HealthReport* health) const override {
    for (const WsConnectionConfig& config : connection_configs_) {
      for (int ind = 0; ind < indent * 4; ind++) {
        health->report += ' ';
      }
      health->report += config.topic + " " + config.url.name + " (fix)\n";
    }
  }

  void CleanUp() override {
    sub_running_ = false;
    for (auto& con : connections_) {
      con->Stop();
    }
    connections_.clear();
  }

 private:
  // connection per symbol group config
  const FeedSubscriptionSpec& spec_;
  std::vector<WsConnectionConfig> connection_configs_;
  std::vector<std::shared_ptr<base::feed::FixConnection>> connections_;
};

}  // namespace coin2::exchange::base::feed
